#ifndef SAFE_CHANNEL
#define SAFE_CHANNEL
#include <deque>
//#include <boost/thread/mutex.hpp>
#include <boost/thread/condition.hpp>
//#include <boost/thread/thread.hpp>
#include <boost/thread.hpp>
#include <exception>
#include <string>

namespace utility
{
template <typename _Tp, typename _queueTp = std::deque<_Tp> >
class safe_queue : private _queueTp
{
private:
    typedef boost::mutex mutex;
    typedef boost::condition condition;
    typedef boost::mutex::scoped_lock lock;
private:
    size_t _max_size;
    mutex _monitor;
    condition _buffer_not_full, _buffer_not_empty;
    volatile bool _may_stop;
public:
    explicit safe_queue(size_t limit = (size_t)-1) : _max_size(limit < 1 ? 1 : limit), _may_stop(false) {}

    safe_queue(_queueTp& queue) 
        : _queueTp(queue), _max_size(queue.max_size()) {}

    virtual ~safe_queue(){}

    // for consumer thread...
    _Tp poll()
    {
        lock lk(_monitor);
        while (!_may_stop && 0 == ((_queueTp *)this)->size())
        {
            _buffer_not_empty.wait(lk);
        }
        // check if caller intentionally calls for stop
        if(_may_stop && 0 == ((_queueTp *)this)->size())
        {
            throw std::logic_error("consume to end");
        }
        // pop back
        _Tp item = pop();
        _buffer_not_full.notify_one();
        return item;
    }
    // for producer thread...
    bool offer(_Tp item)
    {
        lock lk(_monitor);
        while (_max_size == ((_queueTp *)this)->size())
        {
            _buffer_not_full.wait(lk);
        }
        // push front
        push(item);
        _buffer_not_empty.notify_one();
        return true;
    }
    virtual void may_stop(bool bMayStop = true)
    {
        lock lk(_monitor);
        _may_stop = bMayStop;
        if(bMayStop)
        {
            // if producers exit, wake up the
            // waiting consumer threads...
            _buffer_not_empty.notify_all();
        }
    }

    // for outside callers only, calling this on Linux 7.2 within 
    // the above poll() or offer() will deadlock 
    typename _queueTp::size_type size() 
    {
        lock lk(_monitor);
        return ((_queueTp *)this)->size();
    }

protected:
    virtual _Tp pop() // retrieve from end
    {
        _Tp item = ((_queueTp *)this)->back();
        ((_queueTp *)this)->pop_back(); 
        return item;
    }

    virtual void push(const _Tp item) // insert at head...
    {
        ((_queueTp *)this)->insert(_queueTp::begin(), item);
    }
};
}


#endif  // endif of SAFE_CHANNEL

